package com.lg.distributed.transaction.message.retry;

import com.lg.distributed.transaction.message.constant.MQConstants;
import com.lg.distributed.transaction.message.coordinator.DBCoordinator;
import com.lg.distributed.transaction.message.model.RabbitMetaMessage;
import com.lg.distributed.transaction.message.sender.AlertSender;
import com.lg.distributed.transaction.message.sender.RabbitSender;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.util.CollectionUtils;

import java.util.Collections;
import java.util.List;

/**
 * 任务的超策略
 *
 * @author by Mr. Li 2020/12/12 15:13
 */
@Component
@Slf4j
public class ResendProcess {

    @Autowired
    private DBCoordinator dbCoordinator;

    @Autowired
    private AlertSender alertSender;

    @Autowired
    private RabbitSender rabbitSender;

    /**
     * 所有prepare状态的消息，如果超出重试次数，进行拒绝策略，
     * 告警通知
     */
    public void alertPrepareMsg() throws Exception {
        List<RabbitMetaMessage> msgPrepare = dbCoordinator.getMsgPrepare();
        msgPrepare.forEach(message -> alertSender.doSomething(message.getMessageEntity().getMessageId()));
    }

    /**
     * 重发所有 ready状态，并且retry 超过次数的消息
     */
    public void resendReadyMsg() throws Exception {
        List<RabbitMetaMessage> msgReady = dbCoordinator.getMsgReady();
        if (!CollectionUtils.isEmpty(msgReady)) {
            msgReady.forEach(rabbitMetaMessage -> {
                String msgId = rabbitMetaMessage.getMessageEntity().getMessageId();
                // 获取重试次数
                Long retryCount = dbCoordinator.getResendValue(MQConstants.MQ_RESEND_COUNTER, msgId);
                if (retryCount > MQConstants.MAX_CONSUMER_COUNT) { // 在死信队列里面也没能消费掉消息，就走最后的自定义策略了。
                    log.warn("警告，存在无法处理的消息，需要进行告警操作：{}", msgId);
                    alertSender.doSomething(msgId);
                } else {
                    // retry
                    log.info("{} 正在尝试进行重试发送", msgId);
                    // 进入到死信队列
                    rabbitSender.send2DeadQueue(rabbitMetaMessage);
                    dbCoordinator.incrResendKey(MQConstants.MQ_RESEND_COUNTER, msgId);
                }
            });
        }

    }

    /**
     * 重发所有 ready状态，并且retry 超过次数的消息
     */
    public void finallyResendReadyMsg() throws Exception {
        List<RabbitMetaMessage> msgReady = dbCoordinator.getMsgReady();
        if (!CollectionUtils.isEmpty(msgReady)) {
            msgReady.forEach(rabbitMetaMessage -> {
                String msgId = rabbitMetaMessage.getMessageEntity().getMessageId();
                // 获取重试次数
                Long retryCount = dbCoordinator.getResendValue(MQConstants.MQ_RESEND_COUNTER, msgId);
                if (retryCount > MQConstants.MAX_CONSUMER_COUNT) { // 在死信队列里面也没能消费掉消息，就走最后的自定义策略了。
                    log.warn("警告，存在无法处理的消息，需要进行告警操作：{}", msgId);
                    alertSender.doSomething(msgId);
                } else {
                    // retry
                    log.info("{} 正在尝试进行重试发送", msgId);
                    // 进入到死信队列
                    rabbitSender.send(rabbitMetaMessage);
                    dbCoordinator.incrResendKey(MQConstants.MQ_RESEND_COUNTER, msgId);
                }
            });
        }

    }
}
